package com.lagou.mq;

import com.lagou.db.Product;
import com.lagou.es.EsService;
import com.lagou.mapper.ProductMapper;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.LongDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;

import java.util.Collections;
import java.util.Optional;
import java.util.Properties;

@Component
public class Consumer {

    @Autowired
    private ProductMapper productMapper;

    @Autowired
    private EsService service;

    @KafkaListener(topics = {"${kafka.topic.name}"})
    public void listener(ConsumerRecord<String, Long> record){
        Optional<Long> value = Optional.ofNullable(record.value());


        if(value.isPresent()){
            Long productId = value.get();

            System.out.println("收到"+productId);

            Product product = productMapper.selectById(productId);

            service.upload(Product.converter(product));


        }
    }



}
